import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.*;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.ArrayList;

public class KuduConnect {
    //
    private KuduClient kuduClient;
    //kudu master地址
    private String KuduMaster;
    //kudu中的表名
    private String kuduTableName = "Student";

    //todo 初始化方法 用于和kudu集群建立连接

    @Before
    public void init() {
        //指定kudu集群master地址
        KuduMaster = "node-1:7051,node-2:7051,node-3:7051";
        KuduClient.KuduClientBuilder kuduClientBuilder = new KuduClient.KuduClientBuilder (KuduMaster);
        //指定客户端和kudu集群socker 超时时间
        kuduClientBuilder.defaultSocketReadTimeoutMs (10000);

        //通过builder中的build方法创建kuduclient
        kuduClient = kuduClientBuilder.build ();
    }

    /**
     * todo 创建表
     */
    @Test
    public void createTable() throws KuduException {
        //判断table是否存在
        if (!kuduClient.tableExists (kuduTableName)) {
            //指定表的schema信息
            ArrayList<ColumnSchema> columnSchemas = new ArrayList<ColumnSchema> ();
            //添加字段的schema信息
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("id", Type.INT32).key (true).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("name", Type.STRING).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("age", Type.INT32).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("sex", Type.INT32).build ());
            Schema schema = new Schema (columnSchemas);

            //指定表的option属性
            CreateTableOptions createTableOptions = new CreateTableOptions ();
            //指定表的分区规则，采用hash分区，根据哈希划分到指定的6个部分中
            ArrayList<String> partitionList = new ArrayList<String> ();
            partitionList.add ("id");
            createTableOptions.addHashPartitions (partitionList, 6);//id hashcode % 6


            //如果表不存在，进行创建表的操作(表名，表的schema，表的属性信息)
            kuduClient.createTable (kuduTableName, schema, createTableOptions);
        }
    }

    /**
     * todo 创建hash分区表
     */
    @Test
    public void createTable_hash() throws KuduException {
        //判断table是否存在
        if (!kuduClient.tableExists ("t_hash_partition")) {
            //指定表的schema信息
            ArrayList<ColumnSchema> columnSchemas = new ArrayList<ColumnSchema> ();
            //添加字段的schema信息
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("id", Type.INT32).key (true).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("name", Type.STRING).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("age", Type.INT32).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("sex", Type.INT32).build ());
            Schema schema = new Schema (columnSchemas);

            //指定表的option属性
            CreateTableOptions createTableOptions = new CreateTableOptions ();
            //指定分区字段 todo id
            ArrayList<String> partitionList = new ArrayList<String> ();
            partitionList.add ("id");
            //指定表的分区规则，采用hash分区，根据哈希划分到指定的6个部分中
            createTableOptions.addHashPartitions (partitionList, 6);//id hashcode % 6


            //如果表不存在，进行创建表的操作(表名，表的schema，表的属性信息)
            kuduClient.createTable ("t_hash_partition", schema, createTableOptions);
        }
    }

    /**
     * todo 创建range分区表
     */
    @Test
    public void createTable_range() throws KuduException {
        //判断table是否存在
        if (!kuduClient.tableExists ("t_range_partition")) {
            //指定表的schema信息
            ArrayList<ColumnSchema> columnSchemas = new ArrayList<ColumnSchema> ();
            //添加字段的schema信息
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("id", Type.INT32).key (true).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("name", Type.STRING).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("age", Type.INT32).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("sex", Type.INT32).build ());
            Schema schema = new Schema (columnSchemas);

            //指定表的option属性
            CreateTableOptions tableOptions = new CreateTableOptions ();
            //指定分区字段 todo id
            ArrayList<String> partitionList = new ArrayList<String> ();
            partitionList.add ("id");
            //指定表的分区规则，采用hash分区，根据哈希划分到指定的6个部分中
            //createTableOptions.addHashPartitions (partitionList, 6);//id hashcode % 6

            //指定range分区字段
            tableOptions.setRangePartitionColumns (partitionList);
            //指定range策略
            /**
             * 0 <= id < 10
             * 10 <= id < 20
             * 20 <= id < 30
             * 30 <= id < 40
             * 40 <= id < 50
             */
            int count = 0;
            for (int i = 0; i <= 5; i++) {
                //指定range的下界
                PartialRow lower = schema.newPartialRow ();
                lower.addInt ("id", count);
                count += 10;
                //指定range的上届
                PartialRow upper = schema.newPartialRow ();
                upper.addInt ("id", count);
                tableOptions.addRangePartition (lower, upper);
            }

            //如果表不存在，进行创建表的操作(表名，表的schema，表的属性信息)
            kuduClient.createTable ("t_range_partition", schema, tableOptions);
        }
    }

    /**
     * todo 创建多级分区表
     */
    @Test
    public void createTable_multi() throws KuduException {
        //判断table是否存在
        if (!kuduClient.tableExists ("t_multi_partition")) {
            //指定表的schema信息
            ArrayList<ColumnSchema> columnSchemas = new ArrayList<ColumnSchema> ();
            //添加字段的schema信息
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("id", Type.INT32).key (true).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("name", Type.STRING).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("age", Type.INT32).build ());
            columnSchemas.add (new ColumnSchema.ColumnSchemaBuilder ("sex", Type.INT32).build ());
            Schema schema = new Schema (columnSchemas);

            //指定表的option属性
            CreateTableOptions tableOptions = new CreateTableOptions ();
            //指定分区字段 todo id
            ArrayList<String> partitionList = new ArrayList<String> ();
            partitionList.add ("id");
            //指定表的分区规则，采用hash分区，根据哈希划分到指定的3个部分中
            tableOptions.addHashPartitions (partitionList, 3);//id hashcode % 3

            //指定range分区字段
            tableOptions.setRangePartitionColumns (partitionList);
            //指定range策略
            /**
             * 0 <= id < 10
             * 10 <= id < 20
             * 20 <= id < 30
             * 30 <= id < 40
             * 40 <= id < 50
             */
            int count = 0;
            for (int i = 0; i <= 5; i++) {
                //指定range的下界
                PartialRow lower = schema.newPartialRow ();
                lower.addInt ("id", count);
                count += 10;
                //指定range的上届
                PartialRow upper = schema.newPartialRow ();
                upper.addInt ("id", count);
                tableOptions.addRangePartition (lower, upper);
            }

            //如果表不存在，进行创建表的操作(表名，表的schema，表的属性信息)
            kuduClient.createTable ("t_multi_partition", schema, tableOptions);
        }
    }

    /**
     * todo 删除表
     */
    @Test
    public void dropTable() throws KuduException {
        if (kuduClient.tableExists (kuduTableName)) {

            kuduClient.deleteTable (kuduTableName);

        }

    }

    /**
     * todo 向表插入数据
     *
     * @throws KuduException
     */

    @Test
    public void insertTable() throws KuduException {
        //向表加数据需要一个 kuduSession 对象
        KuduSession kuduSession = kuduClient.newSession ();
        //设置提交数据为自动flush
        kuduSession.setFlushMode (SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);

        //打开本次操作的表名
        KuduTable kuduTable = kuduClient.openTable (kuduTableName);
        //需要使用 kuduTable 来构建 Operotion 的子类实列对象，此处是insert操作
        for (int i = 1; i <= 10; i++) {
            Insert insert = kuduTable.newInsert ();
            PartialRow row = insert.getRow ();
            row.addInt ("id", i);
            row.addString ("name", "zhangsan-" + i);
            row.addInt ("age", 20 + i);
            row.addInt ("sex", i % 2);
            kuduSession.apply (insert);//最后实现执行数据的加载操作
            //按住ctrl+单机点击apply，发现apply是Operation类型，选中它，按住ctrl+alt+B 发现insert是它的子类
        }
    }

    @Test
    public void queryData() throws KuduException {
        //构建一个查询的扫描器(在扫描器中指定需要操作的表名)
        KuduScanner.KuduScannerBuilder kuduScannerBuilder = kuduClient.newScannerBuilder (kuduClient.openTable (kuduTableName));
        //创建集合，用于存储扫描字段的信息
        ArrayList<String> columnsList = new ArrayList<String> ();
        columnsList.add ("id");
        columnsList.add ("name");
        columnsList.add ("age");
        columnsList.add ("sex");
        kuduScannerBuilder.setProjectedColumnNames (columnsList);
        //调用build方法执行数据的扫描，得到返回结果
        KuduScanner kuduScanner = kuduScannerBuilder.build ();
        //遍历
        while (kuduScanner.hasMoreRows ()) {
            RowResultIterator rowResults = kuduScanner.nextRows ();

            while (rowResults.hasNext ()) {
                RowResult row = rowResults.next ();
                int id = row.getInt ("id");
                String name = row.getString ("name");
                int age = row.getInt ("age");
                int sex = row.getInt ("sex");
                System.out.println ("id=" + id + "name=" + name + "age=" + age + "sex=" + sex);
            }
        }

    }

    /**
     * todo 修改表数据
     *
     * @throws KuduException
     */

    @Test
    public void updateData() throws KuduException {
        //向表加数据需要一个 kuduSession 对象
        KuduSession kuduSession = kuduClient.newSession ();
        //设置提交数据为自动flush
        kuduSession.setFlushMode (SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);

        //打开本次操作的表名
        KuduTable kuduTable = kuduClient.openTable (kuduTableName);
        //需要使用 kuduTable 来构建 Operotion 的子类实列对象，此处是update操作
        Update update = kuduTable.newUpdate ();
        //newUpdate：如果添加的主键id不存在，不报错，也不添加进来
        //newUpsert：如果添加的主键id不存在，则会添加进来
        PartialRow row = update.getRow ();
        row.addInt ("id", 1);//如果指定id不存在，不报错
        row.addString ("name", "青果");
        row.addInt ("age", 50);
        row.addInt ("sex", 1);
        kuduSession.apply (update);//最后实现执行数据的修改操作
    }

    /**
     * todo 删除表中数据
     *
     * @throws KuduException
     */

    @Test
    public void deleteData() throws KuduException {
        //向表加数据需要一个 kuduSession 对象
        KuduSession kuduSession = kuduClient.newSession ();
        //设置提交数据为自动flush
        kuduSession.setFlushMode (SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);

        //打开本次操作的表名
        KuduTable kuduTable = kuduClient.openTable (kuduTableName);
        //需要使用 kuduTable 来构建 Operotion 的子类实列对象，此处是insert操作
        Delete delete = kuduTable.newDelete ();
        PartialRow row = delete.getRow ();
        row.addInt ("id", 100);
        kuduSession.apply (delete);//最后实现执行数据的加载操作
    }

    @After
    public void close() throws KuduException {
        //如果客户端未关闭，执行close操作
        if (kuduClient != null) {
            kuduClient.close ();
        }
    }


}
